// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_SESSION_HPP_
#define _XBASE_SESSION_HPP_

#include <boost/intrusive_ptr.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/system/error_code.hpp>
#include <boost/asio/ip/tcp.hpp>

#include <map>
#include <queue>
#include <utility>

#include "dynamic_buffer.hpp"

#include "async_module_base.hpp"
#include "session.hpp"
#include "protocol.hpp"

namespace xbase
{
	class session
		: public async_module_base<io_service_pool::xbase_level, io_service_pool::socket_level>
	{
	public:
		typedef async_module_base<io_service_pool::xbase_level, io_service_pool::socket_level> base_type;
		typedef session this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef boost::function<void()> on_closed_handler_type;

		typedef boost::system::error_code error_code;

		typedef protocol::packet_ptr packet_ptr;

		typedef boost::function<void(packet_ptr, const error_code &)> handler_type;

		typedef std::map<std::size_t, handler_type> handler_map_type;

		typedef std::pair<packet_ptr, handler_type> item_type;

		typedef boost::asio::ip::tcp::socket socket_type;

	public:
		session(io_service_pool &pool)
			: base_type(pool, "session")
			, m_socket(this->get_lower_level_service())
			, m_async_read_actived(false)
			, m_read_buffer()
			, m_request_queue()
			, m_handler_map() {}

			~session()
			{
				close();
			}

			socket_type &socket()
			{
				return m_socket;
			}

			// active async read loop
			void start()
			{
				this->get_lower_level_service().dispatch(boost::bind(&this_type::active_async_read,
					this_ptr()));
			}

			void set_on_closed_handler(on_closed_handler_type handler)
			{
				m_on_closed = handler;
			}

			void close()
			{
				boost::mutex::scoped_lock lock(m_mutex);
				if(m_socket.is_open())
				{
					m_socket.close();
					lock.unlock();
					if(m_on_closed)
						this->get_upper_level_service().dispatch(m_on_closed);
				}
			}

			void async_write(packet_ptr packet, handler_type handler)
			{
				item_type item(packet, handler);

				boost::mutex::scoped_lock lock(m_mutex);
				const bool not_actived = m_request_queue.empty();
				m_request_queue.push(item);
				lock.unlock();

				if(not_actived)
				{
					// 向下层服务发起异步写请求
					this->get_lower_level_service().dispatch(boost::bind(&this_type::active_async_write,
						this_ptr()));
				}
			}

	protected:
		std::size_t get_packet_id() {
			return m_packet_id.get();
		}

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		// 
		// 在lower_level service上运行的方法
		//

		void active_async_read();

		void active_async_write();

		void on_written(const error_code &ec,
			std::size_t bytes);

		void on_read(const error_code &e,
			std::size_t bytes_transfered);

		void dispatch_error(const error_code &ec);

		// 根据id寻找合适的consumer handler，来处理[begin,end)中的协议数据；
		// 如果id找不到，则调用用户指定的default_consumer handler处理。
		void dispatch(packet_ptr p, const error_code &ec);

	private:
		socket_type m_socket;
		uuid<this_type> m_packet_id;

		bool m_async_read_actived;
		simple_buffer m_read_buffer;
		simple_buffer m_write_buffer;

		std::queue<item_type> m_request_queue;
		boost::mutex m_mutex;

		handler_map_type m_handler_map;
		on_closed_handler_type m_on_closed; // 在upper_level中执行
	};
	
	typedef session::ptr_type session_ptr;

} // namespace xbase

#endif	// _XBASE_SESSION_HPP_
